package edu.kit.ksri.db.mturk.simulations.performance;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Map;
import java.util.Queue;
import java.util.StringTokenizer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.log4j.Logger;

import com.amazonaws.mturk.requester.Assignment;

/**
 * Copyright (c) 2010-2011 Karlsruhe Institute of Technology (KIT), Germany
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the "Software"),
 * to deal in the Software without restriction, including without limitation
 * the rights to use, copy, modify, merge, publish, distribute, sublicense,
 * and/or sell copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 * DEALINGS IN THE SOFTWARE.
 * 
 * --------------------------------------------------------------------------
 * 
 * Software: CSP/WMV tool for dynamic quality management of microtasks
 * http://www.ksri.kit.edu/Upload/Users/PeopleClouds/HCOMP2011_An_extendable_toolkit_for_managing_quality_of_human-based_eServices.pdf
 * 
 * Organization: Karlsruhe Institute of Technology (KIT), Germany
 * http://www.kit.edu
 * 
 * Project: PeopleClouds project of the Karlsruhe Service Research Institute(KSRI)
 * http://www.ksri.kit.edu/Default.aspx?PageId=818
 * 
 * Contributing authors: David Bermbach, Sandra Rath, Pascal Wichmann, Robert Kern
 */
public class PerfSimMTurk {

	private static final Logger basiclog = Logger.getLogger("basic." + PerfSimMTurk.class);
	
	/** all hit ids */
	private static String[] hitIds;

	/** all results which can be retrieved */
	private static final ConcurrentHashMap<Integer, Queue<Assignment>> outputAnswers = new ConcurrentHashMap<Integer, Queue<Assignment>>();

	/**
	 * all question ids, the largest question id number must be equal to
	 * {@link PerfSimWorker}.possibleAnswers.length -1
	 */
	private static final Queue<Integer> questionIds = new ConcurrentLinkedQueue<Integer>();

	/** contains all qualification values for workers */
	private static final Map<String, Integer> newQualificationValues = new ConcurrentHashMap<String, Integer>();

	/** determines the end of the simulation */
	private static boolean running = true;

	/** references to all {@link PerfSimWorker} {@link Thread}s */
	private static final ArrayList<Thread> workerThreads = new ArrayList<Thread>();

	/** counts the number of added answers */
	private static int counter = 0;

	/**
	 * initializes and starts the {@link PerfSimMTurk}
	 * 
	 * @param file
	 *            filename where the input data for the workers can be found.
	 *            must be a comma-separated text file of the form worker_id
	 *            qualification_value + answers (all comma-separated) every
	 *            worker should have the same number of possible answers. This
	 *            method makes no safety checks and will crash if any condition
	 *            is violated. # marks a comment line.
	 * 
	 */
	public static void initialize(String file) throws Exception {
		File f = new File(file);
		BufferedReader br = new BufferedReader(new FileReader(f));
		String line = br.readLine();
		int nrOfAnswers = 0;
		while (line != null) {
			if (line.startsWith("#"))
				continue;
			StringTokenizer st = new StringTokenizer(line, ",");
			String[] tokens = new String[st.countTokens() - 2];
			String workerId = null, qualificationValue = null;
			int counter = -2;
			while (st.hasMoreTokens()) {
				if (counter == -2) {
					workerId = st.nextToken();
					counter++;
				} else if (counter == -1) {
					qualificationValue = st.nextToken();
					counter++;
				} else
					tokens[counter++] = st.nextToken();
			}
			PerfSimWorker sw = new PerfSimWorker(workerId, tokens, Integer
					.parseInt(qualificationValue));
			workerThreads.add(new Thread(sw));
			nrOfAnswers = tokens.length;
			line = br.readLine();
		}
		hitIds = new String[nrOfAnswers];
		for (int i = 0; i < nrOfAnswers; i++) {
			hitIds[i] = "" + i;
		}
		running = true;
		for (Thread t : workerThreads)
			t.start();
	}

	/**
	 * @return the running
	 */
	public static boolean isRunning() {
		return running;
	}

	/**
	 * @param e
	 * @return
	 * @see java.util.Queue#add(java.lang.Object)
	 */
	public static boolean addHIT(Integer e) {
		basiclog.info("Adding hit number " + e);
		outputAnswers.put(e, new ConcurrentLinkedQueue<Assignment>());
		return questionIds.add(e);
	}

	public static boolean extendHIT(Integer e) {
		try {
			synchronized (questionIds) {
				ArrayList<Integer> temp = new ArrayList<Integer>();
				while (questionIds.size() > 0) {
					temp.add(questionIds.remove());
				}
				questionIds.add(e);
				for (Integer ee : temp) {
					questionIds.add(ee);
				}
				if(!outputAnswers.containsKey(e)){
					outputAnswers.put(e, new ConcurrentLinkedQueue<Assignment>());
				}
				basiclog.info("HIT "+e+" extended.");
				return true;
			}
		} catch (Exception e1) {
			e1.printStackTrace();
			basiclog.warn("HIT "+e+" not extended.");
			return false;
		}
	}

	/**
	 * @return
	 * @see java.util.Collection#isEmpty()
	 */
	public static boolean hasHITs() {
		return questionIds.isEmpty();
	}

	/**
	 * @return
	 * @see java.util.Queue#remove()
	 */
	public static Integer getNextHIT() {
		Integer temp = questionIds.remove();
		basiclog.info("Retrieving hit number " + temp + " (further "
				+ questionIds.size() + " HITs exist)");
		return temp;
	}

	/**
	 * @param e
	 * @return
	 * @see java.util.Queue#add(java.lang.Object)
	 */
	public static boolean addAnswer(String e, int hitId, String workerId,
			Calendar acceptTime) {
		counter++;
		basiclog.info("Adding answer " + e + " for HIT " + hitId + " (" + counter
				+ " answers in total).");
		return outputAnswers.get(hitId).add(
				createAssignment(e, hitId, workerId, acceptTime));
	}

	/**
	 * @return
	 * @see java.util.Collection#isEmpty()
	 */
	public static boolean hasAnswers() {
		return outputAnswers.isEmpty();
	}

	/**
	 * @return
	 * @see java.util.Collection#remove()
	 */
	public static Assignment getNextAnswerForHit(int hitId) {
		return outputAnswers.get(hitId).remove();
	}

	/**
	 * @param key
	 * @param value
	 * @return
	 * @see java.util.Map#put(java.lang.Object, java.lang.Object)
	 */
	public static Integer addQualificationUpdate(String key, Integer value) {
		return newQualificationValues.put(key, value);
	}

	/**
	 * @param key
	 * @return
	 * @see java.util.Map#remove(java.lang.Object)
	 */
	public static Integer getQualificationUpdate(Object key) {
		return newQualificationValues.remove(key);
	}

	private static Assignment createAssignment(String answerValue, int hitid,
			String workerId, Calendar acceptTime) {
		Assignment a = new Assignment();
		a.setAnswer(buildXMLAnswer(answerValue));
		a.setHITId("" + hitid);
		a.setSubmitTime(Calendar.getInstance());
		a.setWorkerId(workerId);
		a.setAcceptTime(acceptTime);
		a.setAssignmentId("assignment_" + hitid + "_by_" + workerId);
		return a;
	}

	private static String buildXMLAnswer(String answerValue) {
		return "<QuestionFormAnswers xmlns=\"http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd\">"
				+ "<Answer>    <QuestionIdentifier>Dummy Question</QuestionIdentifier>    <FreeText>"
				+ answerValue
				+ "</FreeText>  </Answer></QuestionFormAnswers>";
	}

	/**
	 * @return the hitIds
	 */
	public static String[] getHitIds() {
		return hitIds;
	}
}
